package cn.tedu.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 编写一个方法作为生产端代码
 * 编写一个方法作为消费端代码
 * 在rabbitmq创建对应的组件 queue，实现一发一接的简单模式
 */
public class WorkMode {
    /**
     * 实现生产端或消费端之前，创建连接对象
     */
    Channel channel;
    String ip = "10.42.182.184";
    int port = 5672;
    String userName = "guest";
    String password = "guest";

    @Before
    public void initChannel() {
        // 获取连接工厂对象，赋值连接信息 ip/port/userpassword/virtual host
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(ip);
        factory.setPort(port);
        // 默认值
//        factory.setUsername(userName);
//        factory.setPassword(password);
//        factory.setVirtualHost("/");
        try {
            Connection connection = factory.newConnection();
            channel = connection.createChannel();
        } catch (TimeoutException timeoutException) {
            timeoutException.printStackTrace();
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }

    /**
     * 生产端逻辑
     */
    @Test
    public void send() {
        // 准备一个即将发送的消息字符串
        String msg = "小红情书: 我就喜欢你每次给我送的大白薯";
        try {
            // 发送之前，声明一个队列
            /**
             * queue: String 队列名称
             * durable: Boolean 表示队列是否支持持久化
             * exclusive: Boolean 表示队列是否专属，专属于当前连接对象Connection. true 表示操作连接队列只有创建connection才能进行
             * autoDelete: Boolean 表示是否自动删除。最后一个channel连接使用完队列是否删除 true 删除 false 不删
             * args: Map对象，创建队列的各种属性, 例如，消息的存活最长时间，消息的最大长度
             */
            channel.queueDeclare("小红", false, false, false, null);
            // 队列和交换机属于virtual host分配资源，一个队列绑定virtual host
            // 在同一批资源中，队列名称不相同，声明时队列存在则声明；不存在在创建
            // channel连接对象的API将消息发送出去
            /**
             * 方法参数
             * exchange: String 类型 代表交换机名称
             * routingKey: String 路由key 默认交换机中就是队列名称值
             * props: BasicProperties, 表示消息的属性
             * bytes: byte[] 消息体的二进制数据
             */
            for (int i = 0; i < 100; ++i)
                channel.basicPublish("", "小红", null, (msg + ": " + i).getBytes());
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }

    /**
     * 消费端逻辑
     */
    @Test
    public void consumer01() {
        // 创建消费端对象
        QueueingConsumer consumer = new QueueingConsumer(channel);
        try {
            // consumer绑定channel后就具备了绑定queue的能力
            // 绑定消费对象和队列"小红"，实现一对一监听
            /**
             * autoAck: Boolean 自动确认回执
             * 回执: 告诉rabbitmq 是否消费消息成功，返回true，rabbitmq将会把该消息删除，返回false，该消息继续保存
             */
            channel.basicQos(1);
            channel.basicConsume("小红", false, consumer);
            // 配置客户端消费逻辑，每次空闲时(回执之后)最多处理一条消息
            while (true) {
                // 监听获取消息信息
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();// 从监听队列中取出一条存在的消息
                Thread.sleep(50);
                // delivery接收的除了消息体body以外，还有header properties
                System.out.println("阿强拿到消息: " +  new String(delivery.getBody()));
                // 手动回执
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        } catch (IOException ioException) {
            ioException.printStackTrace();
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
    }

    /**
     * 消费端逻辑
     */
    @Test
    public void consumer02() {
        // 创建消费端对象
        QueueingConsumer consumer = new QueueingConsumer(channel);
        try {
            // consumer绑定channel后就具备了绑定queue的能力
            // 绑定消费对象和队列"小红"，实现一对一监听
            /**
             * autoAck: Boolean 自动确认回执
             * 回执: 告诉rabbitmq 是否消费消息成功，返回true，rabbitmq将会把该消息删除，返回false，该消息继续保存
             */
            channel.basicQos(1);
            channel.basicConsume("小红", false, consumer);
            while (true) {
                // 监听获取消息信息
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();// 从监听队列中取出一条存在的消息
                Thread.sleep(200);
                // delivery接收的除了消息体body以外，还有header properties
                System.out.println("阿明拿到消息: " +  new String(delivery.getBody()));
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        } catch (IOException ioException) {
            ioException.printStackTrace();
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
    }
}
